package com.talent.web.datasource.influxdb;

import com.influxdb.client.QueryApi;
import com.influxdb.client.WriteApi;
import com.influxdb.client.WriteApiBlocking;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
import com.influxdb.query.FluxTable;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;

import java.time.Instant;
import java.util.HashMap;
import java.util.List;

/**
 * @Author: hszhou
 * @Date: 2025/4/3 9:19
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@Service
@Slf4j
@RequiredArgsConstructor
@ConditionalOnProperty(name = "influx.enabled", havingValue = "true")
public class TlInfluxdbTemplate {


    private final WriteApiBlocking writeApiBlocking;

    private final WriteApi writeApi;

    private final QueryApi queryApi;


    /**
     * 向influxdb中异步写入行协议
     * @param line 具体行协议 一行行协议就是influxDB中的一个数据点
     * 组成
     * measurement,tag_set field_set timestamp
     * 示例
     * weather,location=us-midwest temperature=82,humidity=71 1645425880
     * weather 就是measurement 必有 可以理解成mysql的表
     * location=us-midwest tag 是influxDB中的一个标签 可以有多组 如果有多组就是location=us-midwest,location=us-west中间没有空格
     * 空格  用于分离measurement和filed_set
     * weather temperature=82,humidity=71 1645425880
     */
    public void writeDataByLineProtocolAsync(String line){
        // 时间精度 行协议
        writeApi.writeRecord(WritePrecision.MS,line);
        writeApi.flush();
    }


    /**
     * 向influxdb中同步写入行协议
     * @param line 行协议
     */
    public void writeDataByLineProtocol(String line){
        // 时间精度 行协议
        writeApiBlocking.writeRecord(WritePrecision.MS,line);
    }

    /**
     * 将map同步写入到influxdb中
     * @param measurementName measurement
     * @param tags 标签
     * @param fields 字段值
     */
    public void writePointMap(String measurementName,
                           HashMap<String,String> tags,
                           HashMap<String,Object> fields){
        Point point = Point.measurement(measurementName)
                .addTags(tags)
                .time(Instant.now(),WritePrecision.MS)
                .addFields(fields);

        //转换成行协议
       String lineProtocol = point.toLineProtocol();
       log.info("lineProtocol【{}】",lineProtocol);
       writeApiBlocking.writePoint(point);

    }

    /**
     * 将map异步写入到influxdb中
     * @param measurementName measurement
     * @param tags 标签map
     * @param fields 字段map
     */
    public void writePointMapAsync(String measurementName,
                                HashMap<String,String> tags,
                                HashMap<String,Object> fields){
        Point point = Point.measurement(measurementName)
                .addTags(tags)
                .time(Instant.now(),WritePrecision.MS)
                .addFields(fields);

        //转换成行协议
        String lineProtocol = point.toLineProtocol();
        log.info("lineProtocol【{}】",lineProtocol);
        writeApi.writePoint(point);
        //writeApi.flush();

    }

    /**
     * 同步写入java对象
     * @param object 对象
     * @param writePrecision 时间
     */
    public void writePojo(Object object,WritePrecision writePrecision){
        writeApiBlocking.writeMeasurement(writePrecision,object);
    }

    /**
     * 异步吸入java对象
     * @param object 对象
     * @param writePrecision 时间
     */
    public void writePojoAsync(Object object,WritePrecision writePrecision){

        writeApi.writeMeasurement(writePrecision,object);
    }


    /**
     * 查找数据
     * @param fluxSql fluxSql
     * @return 返回值
     */
    public String  queryRaw(String fluxSql){
        return queryApi.queryRaw(fluxSql);
    }

    /**
     * 查找数据
     * @param fluxSql fluxSql
     * @return 列表
     */
    public List<FluxTable> queryFluxTable(String fluxSql){
        return queryApi.query(fluxSql);
    }


   /**
     * 获取指定时间段内表的数据
     * @param hour 小时
     * @return 列表
     */
    public List<FluxTable> getList(String bucket, String measurement, int hour) {
        StringBuffer buffer = new StringBuffer();
        appendBucketFlux(buffer, bucket);
        appendTimeRangeLastFlux(buffer, hour, "h");
        if (measurement != null) {
            appendTableFlux(buffer, measurement);
        }
        return queryApi.query(buffer.toString());
    }

    /**
     * 获取表数据(时间筛选)
     *
     * @param measurement 表
     * @return 列表
     */
    public List<FluxTable> getList(String bucket, String measurement, String start, String stop) {
        StringBuffer buffer = new StringBuffer();
        appendBucketFlux(buffer, bucket);
        appendTimeRangeFlux(buffer, start, stop);
        appendTableFlux(buffer, measurement);
        return queryApi.query(buffer.toString());
    }


    /**
     * 数据源(桶)表达式
     *
     * @param buffer     buffer
     * @param bucketName 名称
     */
    public static void appendBucketFlux(StringBuffer buffer, String bucketName) {
        buffer.append("from(bucket: \"").append(bucketName).append("\") ");
    }

    /**
     * 表名表达式
     *
     * @param buffer    buffer
     * @param tableName 名称
     */
    public static void appendTableFlux(StringBuffer buffer, String tableName) {
        buffer.append("|> filter(fn: (r) => r._measurement == \"").append(tableName).append("\") ");
    }

    /**
     * 表名表达式
     *
     * @param buffer buffer
     * @param tag    名称
     */
    public static void appendTagFlux(StringBuffer buffer, String tag) {
        buffer.append("|> filter(fn: (r) => r.tag == \"").append(tag).append("\") ");
    }

    /**
     * field表达
     * @param buffer buffer
     * @param field  名称
     */
    public static void appendTagField(StringBuffer buffer, String field) {
        buffer.append("|> filter(fn: (r) => r._field == \"").append(field).append("\") ");
    }


   /**
     * 时间范围表达式  UTC时间
     * @param buffer buffer
     * @param start  开始时间
     * @param stop   结束时间
     */
    public static void appendTimeRangeFlux(StringBuffer buffer, String start, String stop) {
        if (start==null|| start=="") {
            start = "2000-01-01T00:00:00.000Z";
       }
       if (stop==null|| stop=="") {
           buffer.append("|> range(start:").append(start).append(") ");
        } else {
          buffer.append("|> range(start:").append(start).append(", stop:").append(stop).append(") ");
       }
   }

    /**
     * 最近多少时间
     * @param buffer 参数
     * @param time   时间
     * @param unit   时间单位
    */
    public static void appendTimeRangeLastFlux(StringBuffer buffer, int time, String unit) {
        buffer.append("|> range(start: -").append(time).append(unit).append(" )");
    }
}
